/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * @author vjache
 *
 */
public class AffinitiveExecutorService extends AbstractExecutorService {
	
	private static final Log LOG = LogFactory.getFactory().getInstance(AffinitiveExecutorService.class);
	
	private static final int QUEUE_SIZE_PER_WORKER = 100;

	private final Thread[] _workers;
	private final LinkedBlockingQueue<Runnable>[] _workersQueues;

	public AffinitiveExecutorService() {
		this(2*Runtime.getRuntime().availableProcessors());
	}
	/**
	 * 
	 */
	@SuppressWarnings("unchecked")
	public AffinitiveExecutorService(int aPoolSize) {
		_workers = new Thread[aPoolSize];
		_workersQueues = new LinkedBlockingQueue[aPoolSize];
		for(int i=0;i<aPoolSize;i++)
		{
			final LinkedBlockingQueue<Runnable> queue = _workersQueues[i] = new LinkedBlockingQueue<Runnable>(QUEUE_SIZE_PER_WORKER);
			final int j = i;
			_workers[i] = new Thread(new Runnable() {
				@Override
				public void run() {
					Thread.currentThread().setName("ReduceScheduler-"+j);
					while(true)
					{
						final Runnable task;
						try {
							task = queue.poll(100, TimeUnit.MILLISECONDS);
						} catch (InterruptedException e) {
							LOG.warn(String.format("Worker '%s' have been interrupted.", this),e);
							break;
						}
						if(task!=null)
						{
							try
							{
								task.run();
							}
							catch(Throwable t)
							{
								LOG.error(String.format("Task '%s' failed.", task),t);
							}
						}
					}
				}
			});
		}
		
		for(Thread t:_workers)
			t.start();
	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.ExecutorService#awaitTermination(long, java.util.concurrent.TimeUnit)
	 */
	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit)
			throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.ExecutorService#isShutdown()
	 */
	@Override
	public boolean isShutdown() {
		// TODO Auto-generated method stub
		return false;
	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.ExecutorService#isTerminated()
	 */
	@Override
	public boolean isTerminated() {
		// TODO Auto-generated method stub
		return false;
	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.ExecutorService#shutdown()
	 */
	@Override
	public void shutdown() {
		// TODO Auto-generated method stub

	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.ExecutorService#shutdownNow()
	 */
	@Override
	public List<Runnable> shutdownNow() {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see java.util.concurrent.Executor#execute(java.lang.Runnable)
	 */
	@Override
	public void execute(Runnable task) {
		int workerNum = Math.abs(task.hashCode()) % _workers.length;
		try {
			_workersQueues[workerNum].put(task);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

}
